package com.tzg157.fitness.service.realtime.ads;

import cn.hutool.json.JSONUtil;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;

@Service
public class ADSGenderServiceImpl implements Serializable {
    private KafkaProducer<String, String> kafkaProducer;

    @PostConstruct
    public void init(){
        // Kafka 生产者配置
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        kafkaProducer = new KafkaProducer<>(producerProps);
    }

    public void sendToMQ(List<Row> rows) throws StreamingQueryException, TimeoutException {
        /// 将 Dataset<Row> 转换为适合 Kafka 发送的格式
//        Dataset<Row> kafkaDataset = dataset.selectExpr("CAST(gender AS STRING) AS key", "to_json(struct(*)) AS value");
//
//        // 将数据写入 Kafka
//        kafkaDataset.write()
//                .format("kafka")
//                .option("kafka.bootstrap.servers", "localhost:9092")
//                .option("topic", "genderAds")
//                .save();

//        queueMap.put("adsGender", JSONUtil.toJsonStr(rows));
//        System.out.println("*************************** ===>" + queueMap);
        // 等待查询终止
//        query.awaitTermination();
    }
}
